"""
案例分析:完成JSON商品统计
"""
from typing import final

from pyspark import SparkContext,SparkConf
import os
import json
os.environ["PYSPARK_PYTHON"] = "C:/Users/86131/AppData/Local/Programs/Python/Python39/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)

#TODO 需求1: 城市销售额排名
#1.1 读取文件得到RDD
file_rdd = sc.textFile("C:/Users/86131/Desktop/课程/python资料/资料/第15章资料/资料/orders.txt")
# print(file_rdd.collect())
#1.2 取出一个JSON字符串
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
#1.3 将一个JSON字符串转成字典
dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
#1.4 取出城市的销售额
#(城市,销售额)
city_with_money_rdd = dict_rdd.map(lambda x: (x["areaName"], int(x["money"])))
#1.5 对城市的销售额进行聚合
city_result_rdd = city_with_money_rdd.reduceByKey(lambda a, b: a+b)
#1.6 按销售额的聚合结果进行排序
city_final1_rdd = city_result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1 )
#打印
print(f"需求1的结果: {city_final1_rdd.collect()}")

#TODO 需求2: 全部城市有哪些商品类别正在售卖
#2.1取出全部商品类别
category_rdd =  dict_rdd.map(lambda x: x["category"]).distinct()
print(f"需求2的结果: {category_rdd.collect()}")
#2.2对商品类别进行去重

#TODO 需求3: 北京有哪些商品在售卖
#3.1过滤出北京的数据
beijing_data_rdd = dict_rdd.filter(lambda x: x["areaName"] == "北京")
#3.2 取出全部商品类别
result3_rdd = beijing_data_rdd.map(lambda x: x["category"]).distinct()
#3.3对全部商品类型进行去重

print(f"需求3的结果: {result3_rdd.collect()}")
sc.stop()